package org.aksw.xoperator.sparql;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.FutureTask;
import java.util.concurrent.ThreadPoolExecutor;

import org.aksw.xoperator.sparql.http.RemoteSparqlEndpoint;
import org.aksw.xoperator.sparql.local.LocalSparqlEndpoint;
import org.aksw.xoperator.sparql.p2p.P2PFacade;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jivesoftware.smack.SmackConfiguration;

/**
 * coordinates the parlell querying of the different querying means (http/local/p2p)
 */

public class SparqlEndpointFacade extends AbstractThreadPooledSparqler {

	private LocalSparqlEndpoint localSparql;
	
	private P2PFacade p2pSparql;
	
	private RemoteSparqlEndpoint remoteSparql;
		
	Log log = LogFactory.getLog(SparqlEndpointFacade.class);


	public void process(SparqlQuery query) {

		List<SparqlExecutionFuture<SparqlResult>> queries = getFuturesFor(query);
		processFutures(queries,SmackConfiguration.getPacketReplyTimeout());
		
		for (SparqlResult result : query.getResults().values()) {
			if((result.getAnswer()==null||result.getAnswer().isEmpty())&&(result.getErrorMsg()==null||result.getErrorMsg().isEmpty())){
				result.setErrorMsg("Unable to retrieve an response");
			}
		}
	}

	public SparqlEndpointFacade(ThreadPoolExecutor poolExecuter,
			 LocalSparqlEndpoint localSparql, P2PFacade p2pSparql, RemoteSparqlEndpoint remoteSparql) {
		super(poolExecuter);
		this.localSparql = localSparql;
		this.p2pSparql = p2pSparql;
		this.remoteSparql = remoteSparql;
	}

	@Override
	public List<SparqlExecutionFuture<SparqlResult>> getFuturesFor(SparqlQuery query) {
		List<SparqlExecutionFuture<SparqlResult>> queries = new ArrayList<SparqlExecutionFuture<SparqlResult>>();
		//collect all the futures
		queries.addAll(localSparql.getFuturesFor(query));
		queries.addAll(p2pSparql.getFuturesFor(query));
		queries.addAll(remoteSparql.getFuturesFor(query));
		return queries;
	}

}
